package com.lishem.rpc.client.runner;

import com.lishem.common.data.RpcRequest;
import com.lishem.common.data.RpcResponse;
import com.lishem.common.utils.RpcException;
import com.lishem.common.utils.SpringBeanFactory;
import com.lishem.common.utils.StatusEnum;
import com.lishem.rpc.client.cache.ServiceRouteCache;
import com.lishem.rpc.client.channel.ChannelHolder;
import com.lishem.rpc.client.channel.ProviderService;
import com.lishem.rpc.client.cluster.ClusterStrategy;
import com.lishem.rpc.client.cluster.engine.ClusterEngine;
import com.lishem.rpc.client.config.RpcClientConfiguration;
import com.lishem.rpc.client.connector.RpcClientConnector;
import com.lishem.rpc.client.connector.RpcClientInitializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.*;

/**
 * Rpc请求管理器，异步调用实现
 */
public class RpcRequestManager {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcRequestManager.class);

    private static final ConcurrentHashMap<String, ChannelHolder> channelHolderMap = new ConcurrentHashMap<>();

    /**
     * 调用线程池配置
     */
    private static final ExecutorService REQUEST_EXECUTOR = new ThreadPoolExecutor(
            30,
            100,
            0,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(30),
            new BasicThreadFactory.Builder().namingPattern("request-service-connector-%d").build()
    );

    private static RpcRequestPool ZNS_REQUEST_POOL;
    private static ServiceRouteCache SERVICE_ROUTE_CACHE;
    private static String CLUSTER_STRATEGY;

    public static void startRpcRequestManager(RpcRequestPool RpcRequestPool, ServiceRouteCache serviceRouteCache) {
        ZNS_REQUEST_POOL = RpcRequestPool;
        SERVICE_ROUTE_CACHE = serviceRouteCache;
        CLUSTER_STRATEGY = SpringBeanFactory.getBean(RpcClientConfiguration.class).getRpcClientClusterStrategy();
    }

    /**
     * 异步方式发送接口请求调用
     * @param rpcRequest
     * @throws InterruptedException
     * @throws RpcException
     */
    public static void sendRequest(RpcRequest rpcRequest) throws InterruptedException, RpcException {
        // 根据系统配置， 读取对应的负载均衡策略
        ClusterStrategy strategy = ClusterEngine.queryClusterStrategy(CLUSTER_STRATEGY);
        // 根据类信息， 获取服务接口的配置信息
        List<ProviderService> providerServices = SERVICE_ROUTE_CACHE.getServiceRoutes(rpcRequest.getClassName());
        // 根据负载策略， 选取对应的实现引擎
        ProviderService targetServiceProvider = strategy.select(providerServices);

        if (targetServiceProvider != null) {
            String requestId = rpcRequest.getRequestId();
            // 1. 添加计数器
            CountDownLatch latch = new CountDownLatch(1);
            // 2. 加入线程池, 建立连接（连接成功， 会释放计数器）
            REQUEST_EXECUTOR.execute(new RpcClientConnector(requestId, targetServiceProvider, latch));
            // 3. 阻塞等待
            latch.await();
            // 4. 从缓存中获取连接通道
            ChannelHolder channelHolder = channelHolderMap.get(requestId);
            // 5. 发送数据至服务端
            channelHolder.getChannel().writeAndFlush(rpcRequest);
            LOGGER.info("Send request[{}:{}] to service provider successfully", requestId, rpcRequest.toString());
        } else {
            throw new RpcException(StatusEnum.NOT_FOUND_SERVICE_PROVIDER);
        }
    }

    public static void registerChannelHolder(String requestId, ChannelHolder channelHolder) {
        if (StringUtils.isBlank(requestId) || channelHolder == null) {
            return;
        }
        // 缓存记录连接通道
        channelHolderMap.put(requestId, channelHolder);
        LOGGER.info("Register ChannelHolder[{}:{}] successfully", requestId, channelHolder.toString());
        // 缓存记录请求信息， 用于回调结果处理
        ZNS_REQUEST_POOL.submitRequest(requestId, channelHolder.getChannel().eventLoop());
        LOGGER.info("Submit request into RpcRequestPool successfully");
    }

    public static void destroyChannelHolder(String requestId) {
        if (StringUtils.isBlank(requestId)) {
            return;
        }
        ChannelHolder channelHolder = channelHolderMap.remove(requestId);
        try {
            channelHolder.getChannel().closeFuture();
            channelHolder.getEventLoopGroup().shutdownGracefully();
        } catch (Exception ex) {
            LOGGER.error("Close ChannelHolder[{}] error", requestId);
        }

        LOGGER.info("Destroy ChannelHolder[{}] successfully", requestId);
    }
}




/**
 * Rpc请求管理器，非异步调用实现
 */
//@Slf4j
//public class RpcRequestManager {
//
//    // 接口和 提供服务地址的缓存Map
//    private static ServiceRouteCache SERVICE_ROUTE_CACHE;
//
//    // 封装好的Netty处理链路，具体见6.7
//    private static RpcClientInitializer rpcClientInitializer;
//    // 负载均衡策略
//    private static String CLUSTER_STRATEGY;
//
//    private static final ConcurrentHashMap<String, ChannelHolder> channelHolderMap = new ConcurrentHashMap<>();
//
//
//    public static void startRpcRequestManager(ServiceRouteCache serviceRouteCache) {
//        SERVICE_ROUTE_CACHE = serviceRouteCache;
//        rpcClientInitializer = SpringBeanFactory.getBean(RpcClientInitializer.class);
//        CLUSTER_STRATEGY = SpringBeanFactory.getBean(RpcClientConfiguration.class).getRpcClientClusterStrategy();
//    }
//
//    /**
//     * 发送客户端请求，代理对象会调用这个接口，将rpcRequest传入
//     *
//     * @param rpcRequest
//     * @throws InterruptedException
//     * @throws RpcException
//     */
//    public static RpcResponse sendRequest(RpcRequest rpcRequest) throws InterruptedException, RpcException {
//        // 根据系统配置， 读取对应的负载均衡策略
//        ClusterStrategy strategy = ClusterEngine.queryClusterStrategy(CLUSTER_STRATEGY);
//        // 1. 从缓存中获取RPC服务列表信息
//        List<ProviderService> providerServices = SERVICE_ROUTE_CACHE.getServiceRoutes(rpcRequest.getClassName());
//        // 根据负载策略， 选取对应的实现引擎
//        ProviderService targetServiceProvider = strategy.select(providerServices);
//        if (targetServiceProvider != null) {
//            String requestId = rpcRequest.getRequestId();
//            // 3. 发起远程调用
//            RpcResponse response = requestByNetty(rpcRequest, targetServiceProvider);
//            log.info("Send request[{}:{}] to service provider successfully", requestId, rpcRequest.toString());
//            return response;
//        } else {
//            throw new RpcException(StatusEnum.NOT_FOUND_SERVICE_PROVIDER);
//        }
//    }
//
//    /**
//     * 采用Netty进行远程调用
//     */
//    public static RpcResponse requestByNetty(RpcRequest rpcRequest, ProviderService providerService) {
//
//        // 1. 创建Netty连接配置
//        EventLoopGroup worker = new NioEventLoopGroup();
//        Bootstrap bootstrap = new Bootstrap();
//        bootstrap.group(worker)
//                .channel(NioSocketChannel.class)
//                .remoteAddress(providerService.getServerIp(), providerService.getNetworkPort())
//                .handler(rpcClientInitializer);
//        try {
//            // 2. 建立连接
//            ChannelFuture future = bootstrap.connect().sync();
//            if (future.isSuccess()) {
//                ChannelHolder channelHolder = ChannelHolder.builder()
//                        .channel(future.channel())
//                        .eventLoopGroup(worker)
//                        .build();
//                log.info("Construct a connector with service provider[{}:{}] successfully",
//                        providerService.getServerIp(),
//                        providerService.getNetworkPort()
//                );
//
//                // 3. 创建请求回调对象
//                final RequestFuture<RpcResponse> responseFuture = new SyncRequestFuture(rpcRequest.getRequestId());
//                // 4. 将请求回调放置缓存
//                SyncRequestFuture.syncRequest.put(rpcRequest.getRequestId(), responseFuture);
//                // 5. 根据连接通道， 下发请求信息
//                ChannelFuture channelFuture = channelHolder.getChannel().writeAndFlush(rpcRequest);
//                // 6. 建立回调监听
//                channelFuture.addListener(new ChannelFutureListener() {
//                    @Override
//                    public void operationComplete(ChannelFuture future) throws Exception {
//                        // 7. 设置是否成功的标记
//                        responseFuture.setWriteResult(future.isSuccess());
//                        if (!future.isSuccess()) {
//                            // 调用失败，清除连接缓存
//                            SyncRequestFuture.syncRequest.remove(responseFuture.requestId());
//                        }
//                    }
//                });
//                // 8. 阻塞等待3秒
//                RpcResponse result = responseFuture.get(3, TimeUnit.SECONDS);
//                // 9. 移除连接缓存
//                SyncRequestFuture.syncRequest.remove(rpcRequest.getRequestId());
//
//                return result;
//            }
//        } catch (Exception ex) {
//            ex.printStackTrace();
//        }
//        return null;
//    }
//}